package com.sia.springcloud.process.consumer;

import com.sia.springcloud.kafka.DecodeingKafka;
import com.sia.springcloud.process.influxDB.InfluxDbTool;
import com.sia.springcloud.result.DataCollection;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @Author: Sirius
 * @Date: 2022/5/16 15:09
 * @Description:
 **/
@Slf4j
public class TestConsumer implements Runnable {


    // 创建一个Kafka消费者
    //@Resource
    private KafkaConsumer<String, Object> consumer ;

    //一个容器，其中包含特定主题的每个分区的ConsumerRecord列表。
    // 对于Consumer.poll（long）操作返回的每个主题分区，都有一个ConsumerRecord列表。
    private ConsumerRecords<String, Object> msgList;
    private InfluxDbTool influxDbTool;
    private String topicName;
    private static final String GROUPID = "etl";


    public TestConsumer(String topicName,InfluxDbTool influxDbTool) {
        this.topicName = topicName;
        this.influxDbTool = influxDbTool;
        init();

    }

    @Override
    public void run() {
        System.out.println("---------开始消费---------");
        List<String> list = new ArrayList<String>();
        List<Long> list2 = new ArrayList<Long>();

        // 构造一个批量数据的点
        BatchPoints batchPoints = BatchPoints.builder().build();


        long max = 0;
        // 循环执行
        while (true) {

            try {
                // 通过Kafka消费者拉取消息，如果拉取的消息集合不为空
                msgList = consumer.poll(100);

                if (null != msgList) {

                    // 遍历集合中的每条记录
                    // 将记录的值添加到list中
                    // 将记录的偏移量offset添加到list2中
                    for (ConsumerRecord<String, Object> record : msgList) {


                        DataCollection dataCollection =  (DataCollection) record.value();

                        max = Math.max(max,System.currentTimeMillis()-dataCollection.getDate().getTime());
                        if(record.offset()%10000==0){

                            log.info("拉取的消息的延迟为：" +(System.currentTimeMillis()-dataCollection.getDate().getTime()+"  最大延迟为："+max));
                        }
                        if(record.offset()%10000==0){
                            System.out.println("这是拉取的第 "+record.offset()+" 条数据");
                        }

                        // 设备名作为表名，字段名作为tag
                        Point point = Point.measurement(dataCollection.getDeviceName())
                                .addField("collectTime",dataCollection.getDate().getTime())
                                .tag("variableValue",dataCollection.getValue())
                                .tag("variableName",dataCollection.getPointName())
                                .tag("deviceName",dataCollection.getDeviceName())
                                .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS).build();
                        // 将point放入batch中
                        batchPoints.point(point);
                    }
                    // 将数据保存到数据库中
                    influxDbTool.saveInfluxDb(batchPoints,"modbus");

                    // 业务处理成功后，手动提交
                    consumer.commitSync();

                    // 清空三个集合
                    batchPoints.getPoints().clear();
                    list.clear();
                    list2.clear();

                } else {
                    // 如果拉取的集合为空，线程休眠1秒？为什么？
                    Thread.sleep(1000);
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.info(e.getMessage());
            }
        }
    }



        private void init() {
        Properties props = new Properties();

        //kafka消费的的地址
//        props.put("bootstrap.servers", "172.20.3.81:9092");
        props.put("bootstrap.servers", "127.0.0.1:9092");
        //组名 不同组名可以重复消费
        props.put("group.id", GROUPID);
        //是否自动提交
        props.put("enable.auto.commit", "false");
        //超时时间
        props.put("session.timeout.ms", "30000");
        //一次最大拉取的条数
        props.put("max.poll.records", 5000);
//		earliest当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
//		latest
//		当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
//		none
//		topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
        props.put("auto.offset.reset", "earliest");
        //序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", DecodeingKafka.class.getName());
        this.consumer = new KafkaConsumer<String, Object>(props);
        //订阅主题列表topic
        this.consumer.subscribe(Arrays.asList(topicName));

        System.out.println("test初始化!");
    }


}
